package com.shujia.spark.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("state")
    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    /**
     * 统计全局单词的数量
     *
     */

    //读取socket中的数据
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val kvDS: DStream[(String, Int)] = linesDS.flatMap(_.split('.')).map((_, 1))

    /**
     *
     * 有状态算子  -- 使用当前批次的数据加上之前批次的统计结果，返回新的新的结果
     * 有状态的算子需要将计算的状态保存到hdfs中，所以需要设计checkpoint的路径
     *
     * seq：当前批次一个key所有的value
     * option: 之前批次统计结果（状态），如果是第一次统计这个key，option为None
     *
     */
    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey((seq: Seq[Int], option: Option[Int]) => {
      //统计当前批次单词的数量
      val currCount: Int = seq.sum

      //获取之前批次统计的单词的数量
      val count: Int = option.getOrElse(0)
      //计算新的单词的数量并返回
      Option(currCount + count)
    })

    countDS.print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
